package cn.itcast.streaming.task;

import cn.itcast.streaming.entity.ItcastDataObj;
import cn.itcast.streaming.entity.TripModel;
import cn.itcast.streaming.sink.TripDriveToHBaseSink;
import cn.itcast.streaming.utils.JsonParseUtil;
import cn.itcast.streaming.window.function.DriveSampleWindowFunction;
import cn.itcast.streaming.window.function.DriveTripWindowFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.OutputTag;
import org.datanucleus.util.StringUtils;

/**
 * Author itcast
 * Date 2021/9/24 10:16
 * Desc ①驾驶行程采样分析  ②驾驶行程分析
 * 开发步骤：
 * 1.创建流执行环境
 * 2.获取kafka中的数据
 * 3.将json字符串解析成车辆数据对象
 * 4.过滤出正确的数据并且是行程数据 chargeStatus=2或者chargeStatus=3
 * 0x01:停车充电。0x02:行驶充电。0x03:未充电状态。0x04:充电完成。0xFE：异常。0xFF：无效。
 * 5.分配水印机制，设置最大延迟时间 30s
 * 6.超过3分钟的数据，保存到侧输出流，分析一下数据为什么会延迟
 * 7.对车辆数据进行分组，创建会话窗口
 * 8.数据的采样分析
 * 8.1.应用窗口，数据的采样分析
 * 8.2.将分析的采样数据封装成数组，并将其保存到HBase中
 * 9.数据的行程分析
 * 9.1.应用窗口数据，分析低速、中速、高速车辆的soc、行驶里程、油耗、速度、速度切换的次数等数据封装成对象
 * 9.2.将这个对象保存到hbase中
 * 10.执行流环境任务
 */
public class TripDriveTask extends BaseTask {
    public static void main(String[] args) {
        //1.创建流执行环境,参数分析类名称，chk的目录名称
        StreamExecutionEnvironment env = getEnv(TripDriveTask.class.getSimpleName());
        //2.获取kafka中的数据
        DataStreamSource<String> kafkaStream = getKafkaStream(env, "__tripDriver_consumer_", SimpleStringSchema.class);
        //3.将json字符串解析成车辆数据对象
        DataStream<ItcastDataObj> tripDriveStream = kafkaStream.map(JsonParseUtil::parseJsonToObject)
                //4.过滤出正确的数据并且是行程数据 chargeStatus=2或者chargeStatus=3
                .filter(obj -> StringUtils.isEmpty(obj.getErrorData()))
                .filter(obj -> (obj.getChargeStatus() == 2) || (obj.getChargeStatus() == 3));
        //0x01:停车充电。0x02:行驶充电。0x03:未充电状态。0x04:充电完成。0xFE：异常。0xFF：无效。
        //5.分配水印机制，设置最大延迟时间 30s
        SingleOutputStreamOperator<ItcastDataObj> itcastDataObjWatermark = tripDriveStream
                .assignTimestampsAndWatermarks(
                        new BoundedOutOfOrdernessTimestampExtractor<ItcastDataObj>(Time.seconds(30)) {
                            @Override
                            public long extractTimestamp(ItcastDataObj element) {
                                return element.getTerminalTimeStamp();
                            }
                        }
                );
        //6.超过3分钟的数据，保存到侧输出流，分析一下数据为什么会延迟
        OutputTag<ItcastDataObj> maxLatestData =
                new OutputTag<>("maxLatestData", TypeInformation.of(ItcastDataObj.class));
        //7.对车辆数据进行分组，创建会话窗口
        WindowedStream<ItcastDataObj, String, TimeWindow> itcastDataObjWindowStream = itcastDataObjWatermark
                .keyBy(obj -> obj.getVin())
                .window(
                        EventTimeSessionWindows.withGap(Time.minutes(15L))
                )/*.allowedLateness(Time.minutes(3L))
                .sideOutputLateData(maxLatestData)*/;
        //8.数据的采样分析

        //8.1.应用窗口，数据的采样分析
        SingleOutputStreamOperator<String[]> sampleTripDriveStream = itcastDataObjWindowStream
                .apply(new DriveSampleWindowFunction());

        //8.2.将分析的采样数据封装成数组，并将其保存到HBase中
        //sampleTripDriveStream.addSink(new TripSampleToHBaseSink("TRIPDB:trip_sample"));
        //9.数据的行程分析
        //9.1.应用窗口数据，分析低速、中速、高速车辆的soc、行驶里程、油耗、速度、速度切换的次数等数据封装成对象
        SingleOutputStreamOperator<TripModel> tripModelStream = itcastDataObjWindowStream
                .apply(new DriveTripWindowFunction());
        //9.2.将这个对象保存到hbase中
        tripModelStream.addSink(new TripDriveToHBaseSink("TRIPDB:trip_division"));
        //10.执行流环境任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
